OpenEdge Development:<br />Messaging and ESB Publishing, subscribing, and receiving the customer table in a StreamMessage

The procedures example14.p and example15.p (1 of 2) use RAW transfer to publish, subscribe, and receive the customer table in a StreamMessage. The procedure example14.p publishes the customer table in a StreamMessage; each customer record is a bytes item. The procedure example15.p (1 of 2) 15 subscribes and receives the customer table in a StreamMessage; each customer record is a bytes item.

Start a server for the Sports database. Each client must connect to the database in multi-user mode.

Run example15.p (1 of 2) so the subscriber is running before you publish, as shown:

example15.p
/* Receives the customer table in a Stream message. / DEFINE VARIABLE pubsubsession AS HANDLE. DEFINE VARIABLE msgConsumer1 AS HANDLE. DEFINE TEMP-TABLE custt LIKE customer. RUN jms/pubsubsession.p PERSISTENT SET pubsubsession ("-H localhost -S 5162 "). RUN setBrokerURL IN pubsubsession ("localhost:2506"). RUN beginSession IN pubsubsession. RUN createMessageConsumer IN pubsubsession(THIS-PROCEDURE, "messageHandler", OUTPUT msgConsumer1). RUN subscribe IN pubsubsession ( "topic1", ?, / Not a durable subscription / ?, / No message selector / no, / noLocal */ msgConsumer1). RUN startReceiveMessages IN pubsubsession. WAIT-FOR u1 OF THIS-PROCEDURE. FOR EACH custt: DISPLAY custt with 2 column. END. RUN deleteSession IN pubsubsession.
PROCEDURE messageHandler: DEFINE INPUT PARAMETER messageH AS HANDLE NO-UNDO. DEFINE INPUT PARAMETER messageConsumerH AS HANDLE NO-UNDO. DEFINE OUTPUT PARAMETER replyH AS HANDLE NO-UNDO. DEFINE VAR rawCust AS RAW. DO WHILE NOT DYNAMIC-FUNCTION('endOfStream' IN messageH): DYNAMIC-FUNCTION('moveToNext':U IN messageH). rawCust = DYNAMIC-FUNCTION('readBytesToRaw':U IN messageH). RAW-TRANSFER rawCust TO custt. RELEASE custt. END. RUN deleteMessage IN messageH. APPLY "U1" TO THIS-PROCEDURE. END.

example15.p

/* Receives the customer table in a Stream message. */ 
DEFINE VARIABLE pubsubsession AS HANDLE. 
DEFINE VARIABLE msgConsumer1 AS HANDLE. 
DEFINE TEMP-TABLE custt LIKE customer. 
RUN jms/pubsubsession.p PERSISTENT SET pubsubsession  
                                  ("-H localhost -S 5162 "). 
RUN setBrokerURL IN pubsubsession ("localhost:2506"). 
RUN beginSession IN pubsubsession. 
RUN createMessageConsumer IN pubsubsession(THIS-PROCEDURE, 
                                           "messageHandler", 
                                           OUTPUT msgConsumer1). 
RUN subscribe IN pubsubsession (
                  "topic1", 
                  ?,      /* Not a durable subscription */ 
                  ?,      /* No message selector */ 
                  no,     /* noLocal */ 
                  msgConsumer1). 
RUN startReceiveMessages IN pubsubsession. 
WAIT-FOR u1 OF THIS-PROCEDURE. 
FOR EACH custt: 
    DISPLAY custt with 2 column. 
END. 
RUN deleteSession IN pubsubsession.

PROCEDURE messageHandler: 
DEFINE INPUT PARAMETER messageH AS HANDLE NO-UNDO. 
DEFINE INPUT PARAMETER messageConsumerH AS HANDLE NO-UNDO. 
DEFINE OUTPUT PARAMETER replyH AS HANDLE NO-UNDO. 
DEFINE VAR rawCust AS RAW. 
    DO WHILE NOT DYNAMIC-FUNCTION('endOfStream' IN messageH): 
        DYNAMIC-FUNCTION('moveToNext':U IN messageH). 
        rawCust = DYNAMIC-FUNCTION('readBytesToRaw':U IN messageH). 
        RAW-TRANSFER rawCust TO custt. 
        RELEASE custt. 
    END. 
    RUN deleteMessage IN messageH. 
    APPLY "U1" TO THIS-PROCEDURE. 
END.

Run example14.p, as shown:

example14.p
/* Publishes the customer table in a Stream message. */ DEFINE VARIABLE pubsubsession AS HANDLE. DEFINE VARIABLE mesgH AS HANDLE. DEFINE VARIABLE rawCust AS RAW. RUN jms/pubsubsession.p PERSISTENT SET pubsubsession ("-H localhost -S 5162 "). RUN setBrokerURL IN pubsubsession ("localhost:2506"). RUN beginSession IN pubsubsession. RUN createStreamMessage IN pubsubsession (OUTPUT mesgH). FOR EACH customer: RAW-TRANSFER customer TO rawCust. RUN writeBytesFromRaw IN mesgH(rawCust). END. RUN publish IN pubsubsession ("topic1", mesgH, ?, ?, ?). RUN deleteMessage IN mesgH. RUN deleteSession IN pubsubsession.

example14.p

/* Publishes the customer table in a Stream message. */ 
DEFINE VARIABLE pubsubsession AS HANDLE. 
DEFINE VARIABLE mesgH AS HANDLE. 
DEFINE VARIABLE rawCust AS RAW. 
RUN jms/pubsubsession.p PERSISTENT SET pubsubsession  
                                  ("-H localhost -S 5162 "). 
RUN setBrokerURL IN pubsubsession ("localhost:2506"). 
RUN beginSession IN pubsubsession. 
RUN createStreamMessage IN pubsubsession (OUTPUT mesgH). 
FOR EACH customer: 
    RAW-TRANSFER customer TO rawCust. 
    RUN writeBytesFromRaw IN mesgH(rawCust). 
END. 
RUN publish IN pubsubsession ("topic1", mesgH, ?, ?, ?). 
RUN deleteMessage IN mesgH. 
RUN deleteSession IN pubsubsession.